package com.shujia.flink.tf

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object Demo1Map {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.readTextFile("flink/data/students.txt")


    /**
      *
      * map算子  --- 面向函数编程
      *
      */

    //scala 方法

    val clazzDS: DataStream[String] = studentsDS.map(line => {

      println("map============")

      val clazz: String = line.split(",")(4)
      clazz
    })

    //flink 就算最后没有sink 前面的代码也会执行
    //clazzDS.print()


    /**
      * java 的方式, 实现MapFunction 接口，实现map方法  -- 面向接口编程
      *
      */

    //    val javaDS: DataStream[String] = studentsDS.map(new MyMap)


    //匿名内部类
    val javaDS: DataStream[String] = studentsDS.map(new MapFunction[String, String] {
      override def map(value: String): String = {
        value.split(",")(4)
      }
    })


    env.execute()

  }
}


class MyMap extends MapFunction[String, String] {
  /**
    * 每一条数据执行一次
    *
    * @param value 一行数据
    * @return ： 返回数据
    */
  override def map(value: String): String = {
    value.split(",")(4)
  }
}